//package com.jacky.rxjava.v120;
//
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import rx.Observable;
//import rx.functions.Action1;
//import rx.functions.Func1;
//import rx.functions.Func2;
//import rx.subjects.BehaviorSubject;
//import rx.subjects.PublishSubject;
//import rx.subjects.SerializedSubject;
//
//import java.util.concurrent.CountDownLatch;
//import java.util.concurrent.TimeUnit;
//import java.util.function.Consumer;
//
///**
// * 模拟滑动窗口计数
// * Created by albon on 17/6/24.
// */
//public class RollingWindowTest {
//    /**
//     * 两个数字相加，reduce，scan用
//     */
//    public static final Func2<Integer, Integer, Integer> PUBLIC_SUM = (integer, integer2) -> integer + integer2;
//    //跳过第一个数据，因为给了scan一个默认值0，这个值需要跳过，如果不设置就不需要跳过
//    public static final Func1<Observable<Integer>, Observable<Integer>> WINDOW_SUM = window -> window.scan(0, PUBLIC_SUM).skip(1);
//
//    public static final Func1<Observable<Integer>, Observable<Integer>> INNER_BUCKET_SUM = integerObservable -> integerObservable.reduce(0, PUBLIC_SUM);
//
//
//    public static void testWindowSlide() throws InterruptedException {
//        CountDownLatch countDownLatch = new CountDownLatch(1);
//        BehaviorSubject<Integer> behaviorSubject = BehaviorSubject.create();
//        behaviorSubject
//                // 1秒作为一个基本块,横向移动
//                .window(1000, TimeUnit.MILLISECONDS)
//                //将flatMap汇总平铺成一个事件,然后累加成一个Observable<Integer>对象，比如说1s内有10个对象，被累加起来
//                .flatMap(INNER_BUCKET_SUM)
//                //对这个对象2个发送，步长为1
//                .window(2)
//                //对窗口里面的进行求和,用的scan, 每次累加都会打印出来
//                .flatMap(WINDOW_SUM)
//                .subscribe((Integer integer) ->     // 输出统计数据到日志
//                                System.out.println(Thread.currentThread().getName()+" call...."+integer));
//
//        for (int i = 0; i < 1000; i++) {
//            //200ms生产一个数据，
//            behaviorSubject.onNext(1);
//            Thread.sleep(200);
//        }
//        countDownLatch.await();
//    }
//
//    public static void main(String[] args) throws InterruptedException {
//        testWindowSlide();
//    }
//}